package cn.doitedu.demo7;

import cn.doitedu.beans.EventBean;
import cn.doitedu.beans.RuleMeta;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.nio.ByteBuffer;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/14
 * @Desc: 学大数据，上多易教育
 **/
@Slf4j
public class Entrypoint {

    public static void main(String[] args) throws Exception {

        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8989);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        // 构造kafkaSource
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("user-action-log")
                .setGroupId("ggg")
                .setClientIdPrefix("client-001")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();

        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source");

        SingleOutputStreamOperator<EventBean> beanStream = stream.map(json -> JSON.parseObject(json, EventBean.class));

        KeyedStream<EventBean, Long> keyedUserActions = beanStream.keyBy(EventBean::getUser_id);


        /**
         * 实时监控 规则元数据表
         */
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("doit47") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("doit47.rule_meta") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();


        DataStreamSource<String> ruleMetaCdc = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "cdc");

        // 解析cdc读过来的json，封装成： 规则元数据bean
        SingleOutputStreamOperator<RuleMeta> ruleMetaStream = ruleMetaCdc.map(json -> {

            JSONObject jsonObject = JSON.parseObject(json);
            // 取出操作类型
            String op = jsonObject.getString("op");

            JSONObject dataObject;
            if (op.equals("d")) {
                dataObject = jsonObject.getJSONObject("before");
            } else {
                dataObject = jsonObject.getJSONObject("after");
            }

            // 往数据object中填充op字段
            dataObject.put("op", op);

            // 再把数据object转成我们自己的RuleMeta Bean
            RuleMeta ruleMeta = JSON.parseObject(dataObject.toJSONString(), RuleMeta.class);

            byte[] preSelectUsers = ruleMeta.getPre_select_users();

            // 反序列化出bitmap
            Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf();
            if(preSelectUsers != null) {
                bitmap.deserialize(ByteBuffer.wrap(preSelectUsers));

                // 将反序列化好的bitmap设置到ruleMeta对象中
                ruleMeta.setPre_select_users_bitmap(bitmap);

                ruleMeta.setPre_select_users(null);
            }
            return ruleMeta;

        });

        // 广播规则元数据流
        MapStateDescriptor<String, RuleMeta> desc = new MapStateDescriptor<String, RuleMeta>("rule_metas", String.class, RuleMeta.class);
        BroadcastStream<RuleMeta> ruleMetaBroadcastStream = ruleMetaStream.broadcast(desc);


        // 将 用户行为流  连接   规则元数据广播流
        BroadcastConnectedStream<EventBean, RuleMeta> connectedStream = keyedUserActions.connect(ruleMetaBroadcastStream);


        // 对 连接好的两个流，进行统一处理
        SingleOutputStreamOperator<String> resultStream = connectedStream.process(new CoreFunction());


        resultStream.print();

        env.execute("i love you all");


    }


}
